#include "httpflv_server_session.hpp"
#include "core/rtmp_media_sink.hpp"
#include "core/rtmp_media_source.hpp"
// #include "core/media_event.hpp"
#include "http_server_session.hpp"
#include "core/source_manager.hpp"

#include "transformat/rtmp_to_flv.hpp"
#include "core/flv_media_sink.hpp"

#include "spdlog/spdlog.h"
#include "protocol/flv/flv_protocol.hpp"
#include "protocol/flv/flv_tag.hpp"

using namespace mms;
HttpFlvServerSession::HttpFlvServerSession(std::shared_ptr<HttpRequest> http_req, std::shared_ptr<HttpResponse> http_resp) : StreamSession(http_resp->get_worker()),
send_funcs_channel_(get_worker()->get_io_context()),
recv_coroutine_exited_(get_worker()->get_io_context()),
send_coroutine_exited_(get_worker()->get_io_context()),
alive_timeout_timer_(get_worker()->get_io_context()),
alive_checker_coroutine_exited_(get_worker()->get_io_context()) {
    http_request_ = http_req;
    http_response_ = http_resp;
    send_bufs_.reserve(20);
    set_session_type("http-flv");
}

HttpFlvServerSession::~HttpFlvServerSession() {
    spdlog::debug("destroy HttpFlvServerSession");
}

void HttpFlvServerSession::start_send_coroutine() {
    // 创建发送协程
    auto self(shared_from_this());
    send_coroutine_running_ = true;
    boost::asio::co_spawn(worker_->get_io_context(), [this, self]()->boost::asio::awaitable<void> {
        boost::system::error_code ec;
        while (1) {
            auto send_func = co_await send_funcs_channel_.async_receive(boost::asio::redirect_error(boost::asio::use_awaitable, ec));
            if (ec) {
                break;
            }
            update_active_timestamp();
            bool ret = co_await send_func();
            if (!ret) {
                break;
            }
        }
        co_return;
        }, [this, self](std::exception_ptr exp) {
            (void)exp;
            send_coroutine_exited_.close();
            send_coroutine_running_ = false;
            close(true);
            });
}

boost::asio::awaitable<void> HttpFlvServerSession::stop_send_coroutine() {
    if (send_coroutine_running_) {
        if (send_coroutine_exited_.is_open()) {
            boost::system::error_code ec;
            co_await send_coroutine_exited_.async_receive(boost::asio::redirect_error(boost::asio::use_awaitable, ec));
        }
    }
    co_return;
}

void HttpFlvServerSession::start_alive_checker() {
    auto self(shared_from_this());
    alive_checker_coroutine_running_ = true;
    boost::asio::co_spawn(worker_->get_io_context(), [this, self]()->boost::asio::awaitable<void> {
        boost::system::error_code ec;
        while (1) {
            alive_timeout_timer_.expires_from_now(std::chrono::seconds(5));
            co_await alive_timeout_timer_.async_wait(boost::asio::redirect_error(boost::asio::use_awaitable, ec));
            if (ec) {
                co_return;
            }
            int64_t now_ms = Utils::get_current_ms();
           // spdlog::info("session:{} checking alive...", get_session_name());
            if (now_ms - last_active_time_ >= 50000) {//5秒没数据，超时
                close(true);
                co_return;
            }
           // spdlog::info("session:{} is alive", get_session_name());
        }
        co_return;
        }, [this, self](std::exception_ptr exp) {
            (void)exp;
            alive_checker_coroutine_running_ = false;
            alive_checker_coroutine_exited_.close();}
            );
}

void HttpFlvServerSession::update_active_timestamp() {
    last_active_time_ = Utils::get_current_ms();
}

boost::asio::awaitable<void> HttpFlvServerSession::stop_alive_checker() {
    if (alive_checker_coroutine_running_) {
        alive_timeout_timer_.cancel();
        if (alive_checker_coroutine_exited_.is_open()) {
            boost::system::error_code ec;
            co_await alive_checker_coroutine_exited_.async_receive(boost::asio::redirect_error(boost::asio::use_awaitable, ec));
        }
    }
    co_return;
}

void HttpFlvServerSession::service() {
    auto self(std::static_pointer_cast<HttpFlvServerSession>(shared_from_this()));
    boost::asio::co_spawn(worker_->get_io_context(), [this, self]()->boost::asio::awaitable<void> {
        recv_coroutine_running_ = true;
        auto& query_params = http_request_->get_query_params();
        for (auto& p : query_params) {
            set_param(p.first, p.second);
        }
        // http://ip/app/test.flv?domain=test.publish.com
        std::string domain = http_request_->get_query_param("domain");
        if (domain.empty()) {
            domain = http_request_->get_header("Host");
            auto pos = domain.find(":");
            if (pos != std::string::npos) {
                domain = domain.substr(0, pos);
            }
        }

        set_session_info(domain, http_request_->get_path_param("app"), http_request_->get_path_param("stream"));
        // 1.本机查找
        auto source = SourceManager::get_instance().get_source(get_session_name());
        if (!source) {
            //真找不到源流了，应该是没在播
            http_response_->add_header("Content-Type", "video/x-flv");
            http_response_->add_header("Connection", "Close");
            co_await http_response_->write_header(404, "Not Found");
            close(true);
            co_return;
        }

        // 判断是否需要进行桥转换
        std::shared_ptr<MediaBridge> bridge;
        if (source->get_media_type() != "flv") {
            bridge = source->get_or_create_bridge(source->get_media_type() + "-flv", this);
            if (!bridge) {
                http_response_->add_header("Connection", "close");
                if (!(co_await http_response_->write_header(415, "Unsupported Media Type"))) {
                    close(true);
                }
                co_return;
            }
            source = bridge->get_media_source();
            std::string name = std::string("flv") + source->get_domain() + source->get_app()
            +source->get_stream();
            SourceManager::get_instance().add_source_if_not_exist(name,source);
        }

        flv_media_sink_ = std::make_shared<FlvMediaSink>(get_worker());
        // 创建发送协程
        start_send_coroutine();
        start_alive_checker();
        // 收到新的flv消息
        flv_media_sink_->on_flv_tag([this, self](std::vector<std::shared_ptr<FlvTag>>& flv_tags)->boost::asio::awaitable<bool> {
            boost::system::error_code ec;
            if (flv_tags.size() <= 0) {
                co_return true;
            }

            co_await send_funcs_channel_.async_send(boost::system::error_code{}, std::bind(&HttpFlvServerSession::send_flv_tags, this, flv_tags), boost::asio::redirect_error(boost::asio::use_awaitable, ec));
            if (ec) {
                co_return false;
            }
            co_return true;
            });
        // 关闭flv
        flv_media_sink_->on_close([this, self]() {
            close();
            });

        //找到源流，先发http送头部
        http_response_->add_header("Content-Type", "video/x-flv");
        http_response_->add_header("Connection", "Keep-Alive");
        http_response_->add_header("Access-Control-Allow-Origin", "*");
        if (!(co_await http_response_->write_header(200, "OK"))) {
            close(true);
            co_return;
        }
        // 再发flv头部
        uint8_t flv_header[9];
        auto consumed = FlvHeader::encode(flv_header, 9);
        if (!(co_await http_response_->write_data(std::string_view((char*)flv_header, consumed)))) {
            close(true);
            co_return;
        }

        source->add_media_sink(flv_media_sink_);
        co_return;
        }, [this, self](std::exception_ptr exp) {
            (void)exp;
            recv_coroutine_exited_.close();
            recv_coroutine_running_ = false;
            });
}

void HttpFlvServerSession::close(bool close_connection) {//如果是长连接，则不关闭socket
    if (close_connection) {
        http_response_->close();
    }
    close();
}

void HttpFlvServerSession::close() {
    auto self(this->shared_from_this());
    // todo: how to record 404 error to log.
    if (closed_.test_and_set()) {
        return;
    }

    boost::asio::co_spawn(worker_->get_io_context(), [this, self]()->boost::asio::awaitable<void> {
        spdlog::info("closing HttpFlvServerSession...");
        co_await stop_send_coroutine();
        co_await stop_alive_checker();

        if (recv_coroutine_running_) {
            boost::system::error_code ec;
            co_await recv_coroutine_exited_.async_receive(boost::asio::redirect_error(boost::asio::use_awaitable, ec));
        }

        if (flv_media_sink_) {
            auto source = flv_media_sink_->get_source();//SourceManager::get_instance().get_source(get_session_name());
            if (source) {
                source->remove_media_sink(flv_media_sink_);
                flv_media_sink_->close();
            }
            flv_media_sink_->on_close({});
            flv_media_sink_->on_flv_tag({});
            flv_media_sink_ = nullptr;

        }
        spdlog::info("HttpFlvServerSession closed");
        co_return;
        }, boost::asio::detached);
}

boost::asio::awaitable<bool> HttpFlvServerSession::send_flv_tags(std::vector<std::shared_ptr<FlvTag>> tags) {
    int i = 0;
    send_bufs_.clear();
    for (auto& flv_tag : tags) {
        prev_tag_sizes_[i] = htonl(prev_tag_size_);
        send_bufs_.push_back(boost::asio::const_buffer((char*)&prev_tag_sizes_[i], sizeof(int32_t)));
        std::string_view tag_payload = flv_tag->get_raw_data();
        send_bufs_.push_back(boost::asio::const_buffer(tag_payload.data(), tag_payload.size()));
        prev_tag_size_ = tag_payload.size();
        i++;
    }

    if (send_bufs_.size() <= 0) {
        co_return true;
    }

    if (!co_await http_response_->write_data(send_bufs_)) {
        close(true);
        co_return false;
    }
    co_return true;
}